/**
 * Wihidum - Efficient Broker Hierarchy for Publish Subscribe Middleware
 *
 * License      :   Apache License 2.0
 * Home Page    :   http://code.google.com/p/wihidum/
 * Mailing List :   wihidum@googlegroups.com
 */

package lk.mrt.cse.wihidum.storage;

import org.apache.log4j.Logger;
import org.hibernate.*;
import org.hibernate.criterion.Restrictions;

import java.util.*;

public class BrokerSubscriptionDAO
        implements BrokerSubscriptionManager
{
    Session session = null;
    private static final Logger logger = Logger.getLogger (BrokerSubscriptionDAO.class);

    /**
     * Default constructor
     */
    public BrokerSubscriptionDAO ()
    {
        this.session = DataBaseManager.getSessionFactory ().getCurrentSession ();
    }

    public boolean saveBrokerSubscription(BrokerSubscription brokerSubscription,String type)
    {
         try
        {
            Transaction tx = session.beginTransaction ();
            session.save (brokerSubscription);
            tx.commit ();
            logger.info (type+" BrokerSubscription saved successfully");
            return true;
        }
        catch (HibernateException ex)
        {
            logger.error ("Error in saving BrokerSubscription", ex);
            return false;
        }
    }

    public ArrayList<BrokerSubscription> getBrokerSubscriptionsFilteredByTopic(String topic)
    {
        try
        {
            ArrayList<BrokerSubscription> searchedList;

            session.beginTransaction ();
            Criteria criteria = session.createCriteria (BrokerSubscription.class)
                    .add (Restrictions.eq ("topic", topic))
                    .add (Restrictions.isNotNull("fromBrokerAddress"));               
            searchedList = (ArrayList<BrokerSubscription>) criteria.list ();

            if (searchedList == null || searchedList.isEmpty ())
            {
                logger.debug ("No registered BrokerSubscriptions found for topic : " + topic);
                return new ArrayList<BrokerSubscription> ();
            }
            else
            {
                logger.debug (searchedList.size () + " registered BrokerSubscription(s) found for topic : " + topic);
                return searchedList;
            }
        }
        catch (HibernateException ex)
        {
            logger.error ("Error in searching for BrokerSubscriptions", ex);
            return new ArrayList<BrokerSubscription> ();
        }
    }

    public boolean isSubscribedTo(NeighbourBroker neighbourUnit, String filter)
    {
        try
        {
            String toBroker=neighbourUnit.getBrokerID();
            //logger.info("Checking whether already subscribed to"+toBroker+"on the topic"+filter);
            session.beginTransaction();
            Criteria criteria=session.createCriteria(BrokerSubscription.class).add(Restrictions.eq("toBrokerID",toBroker)).add(Restrictions.eq("topic",filter));
            //criteria.setFirstResult(1);
            //List<BrokerSubscription> results=(List<BrokerSubscription>)criteria.list();
            List results=criteria.list();
            if(results.isEmpty())
            {
                //logger.info("Not subscribed before on this topic");
                return false;
            }
            else if(!(results.isEmpty()))
            {
                 //logger.info("already subscribed on this topic");
                 return true;
            }

        }
        catch (HibernateException ex)
        {

        }
        return false;
    }
    public boolean isSentFromPeer(NeighbourBroker peerUnit,String filter)
    {
        boolean sentFromPeer=false;
         try
        {
            String fromBroker=peerUnit.getBrokerID();
            logger.info("Checking whether subscription sent from"+fromBroker+"on the topic"+filter);
            session.beginTransaction();
            Criteria criteria=session.createCriteria(BrokerSubscription.class).add(Restrictions.eq("fromBrokerID",fromBroker)).add(Restrictions.eq("topic",filter));
            //criteria.setFirstResult(1);
            //List<BrokerSubscription> results=(List<BrokerSubscription>)criteria.list();
            List results=criteria.list();
            if(results.isEmpty())
            {
                //logger.info("Not subscribed before on this topic");
                return false;
            }
            else if(!(results.isEmpty()))
            {
                 //logger.info("already subscribed on this topic");
                 return true;
            }

        }
        catch (HibernateException ex)
        {

        }
        return sentFromPeer;
    }
}
